#include "utils/thread/thread_pool.h"
#include "utils/thread/runnable.h"
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <atomic>

namespace base
{

class RunnablePack
{
public:
    RunnablePack(Runnable *runnable, int priority)
        : runnable_(runnable), priority(priority)
    {
    }

    friend bool operator<(const RunnablePack &p1, const RunnablePack &p2)
    {
        return p1.priority < p2.priority;
    }

    Runnable *runnable_;
    int priority;
};

class ThreadPoolData
{
public:
    ThreadPoolData();
    ~ThreadPoolData();
private:
    friend class ThreadPool;
    std::priority_queue<RunnablePack> thread_queue_;//线程任务队列
    int current_thread_count_{0};
    int max_thread_count_{4};
    std::atomic_int active_thread_count_{0};
    std::atomic_bool init_flag_{false};
    std::mutex mutex_;
    std::condition_variable cv;
    std::vector<std::thread> thread_vector_;
    bool finish_thread_{false};

private:
    void startThread();
    void run();
    bool waitForDone(int msecs = -1);
};
ThreadPoolData::ThreadPoolData() = default;

ThreadPoolData::~ThreadPoolData()
{
    waitForDone();
}
void ThreadPoolData::startThread()
{
    init_flag_.store(true);
    if (current_thread_count_ >= max_thread_count_)
    {
        return;
    }
    for (; current_thread_count_ < max_thread_count_; current_thread_count_++)
    {
        auto thread = std::thread(&ThreadPoolData::run, this);
        thread.detach();
        thread_vector_.push_back(std::move(thread));
    }
}
void ThreadPoolData::run()
{
    while (true)
    {
        std::unique_lock<std::mutex> locker(mutex_);
        cv.wait(locker, [this]
        {
            return !thread_queue_.empty() || finish_thread_;
        });
        if (finish_thread_)
        {
            break;
        }
        Runnable *front = thread_queue_.top().runnable_;
        thread_queue_.pop();
        locker.unlock();
        if (nullptr != front)
        {
            active_thread_count_++;
            if (!front->getRunnableName().empty())
            {
                pthread_setname_np(pthread_self(), front->getRunnableName().c_str());
            }
            front->run();
            if (front->autoDelete())
            {
                delete front;
            }
            active_thread_count_--;
        }
    }
}
bool ThreadPoolData::waitForDone(int msecs)
{
    finish_thread_ = true;
    std::unique_lock<std::mutex> locker(mutex_);
    cv.notify_all();
    std::chrono::time_point<std::chrono::high_resolution_clock> begin = std::chrono::high_resolution_clock::now();
    while (active_thread_count_ > 0)
    {
        if (msecs > 0 && std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::high_resolution_clock::now() - begin).count() >= msecs)
        {
            break;
        }
    }
    return thread_queue_.empty() && 0 == active_thread_count_;
}
ThreadPool::ThreadPool()
    : data_(std::make_unique<ThreadPoolData>())
{
}
ThreadPool::~ThreadPool() = default;

void ThreadPool::start(Runnable *runnable, int priority)
{
    if (!data_->init_flag_.load())
    {
        data_->startThread();
    }
    if (!runnable)
    {
        return;
    }
    std::lock_guard<std::mutex> lock_guard(data_->mutex_);
    data_->thread_queue_.emplace(runnable, priority);
    data_->cv.notify_one();
}
void ThreadPool::clear()
{
    std::lock_guard<std::mutex> lock_guard(data_->mutex_);
    while (!data_->thread_queue_.empty())
    {
        const Runnable *front = data_->thread_queue_.top().runnable_;
        data_->thread_queue_.pop();
        if (nullptr != front && front->autoDelete())
        {
            delete front;
        }
    }
}
void ThreadPool::setMaxThreadCount(int max_thread_count)
{
    data_->max_thread_count_ = max_thread_count;
    data_->startThread();
}
int ThreadPool::maxThreadCount()
{
    return data_->max_thread_count_;
}
bool ThreadPool::waitForDone(int msecs)
{
    return data_->waitForDone(msecs);
}
}
